
import java.util.Properties
import bean.Log
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Duration, State, StateSpec, StreamingContext}
import tools.Configs.RedisConfig
import tools.{ReadJson, RedisOperation, TimeTools}

import scala.collection.mutable
import scala.collection.mutable.HashMap
object LogAnalysis {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME","hdfs")
    val sparkConf = new SparkConf().setAppName("test").setMaster("local[4]")
    val streaming = new StreamingContext(sparkConf,Duration(5000))
    streaming.checkpoint("d:/checkpoint")
    val redisPro = new Properties()
    redisPro.setProperty("address","192.168.0.79")
    redisPro.setProperty("port","6379")
    redisPro.setProperty("auth","HTdata")
    redisPro.setProperty("mode","redispool")
    redisPro.setProperty("connectionTimeout","200")
    val logStream = streaming.socketTextStream("192.168.0.83",9999)
    //过滤日志为登陆日志
    /*val lines = streaming.socketTextStream("192.168.0.83",9999).map(x=>{
      val record = ReadJson.readValue(x,classOf[Log])
      (s"$record.tid:$record.uid",record)
    })*/
    val loginStream = logStream.mapPartitions(it=>{
      var list = List[(String,Log)]()
      val moduleMap = new HashMap[String,Int]
      var faildC = 0
      var successC = 0
      it.foreach(x=>{
        val record = ReadJson.readValue(x,classOf[Log])
        moduleMap.put(record.modular,moduleMap.getOrElse(record.modular,0)+1)
        if("登入系统".equals(record.moduleFunction)) {
          if (record.success == 0) {
            list.::= ((s"$record.tid:$record.uid", record))
            successC = successC + 1
          }
          else {
            faildC = faildC + 1
          }
        }
      })
      val jedis = RedisOperation.getInstance(redisPro).getResource()
      val pl = jedis.pipelined()
      val week = TimeTools.getWeek()
      moduleMap.foreach(entry=>{
        pl.hincrBy(RedisConfig.logModuleBaseKey+week,entry._1,entry._2)
      })
      pl.hincrBy(RedisConfig.logLoginStateKey,"success:"+week,successC)
      pl.hincrBy(RedisConfig.logLoginStateKey,"faild:"+week,faildC)
      pl.sync()
      pl.close()
      jedis.close()
      list.iterator
    })

    //判断是否到目前为止本月每周登陆都超过三次
    def detectWeekOfMonth(arr:Array[Int],index:Int): Boolean ={
      for(i<-0 to index-1){
        if(arr(i)!=3){
          return false
        }
      }
      return true
    }
    //判断本周是否登陆超过三次
    def detectThisWeek(times:Int): Boolean ={
      times == 3
    }

    def mapfun = (tidUid:String,value:Option[Log],state:State[(Int,Array[Int],Int)])=>{
      val record = value.get
      val dayInfo = TimeTools.getDayInfo(record.createTime)
      val loginT = TimeTools.formatTs(record.createTime)
      if(state.exists()){
        val s = state.get()
        if(s._1 == loginT){//上次登录时间为今天
          s._2(dayInfo._1-1) = s._2(dayInfo._1-1) + 1
          state.update((s._1,s._2,s._3))
          (record.createTime,false,false,detectWeekOfMonth(s._2,dayInfo._1),detectThisWeek(s._2(dayInfo._1-1)),record.modular)
        }else{//上次登陆时间不是今天
          //判断数据时间大于状态时间否则抛弃该条数据
          var s3 = s._3
          if(loginT > s._1){
            //判断上次登陆时间是否为本年本月,如果不是本月则清除状态数据中<每周登陆次数><本月连续登陆天数>
            if(!TimeTools.isSameMonth(record.createTime,s._1)){
              for(i<-0 to s._2.length){
                s._2(i) = 0
              }
              s3 = 0
            }
            s._2(dayInfo._1-1) = s._2(dayInfo._1-1) + 1
            val dayOfMonth = s3 + 1
            state.update((loginT,s._2,dayOfMonth))
            //本月是否每天至少登陆一次
            var loginEveryDayOfMonth = false
            var loginEveryDayOfWeek = false
            if(dayOfMonth==dayInfo._2)
              loginEveryDayOfMonth = true
            if(loginEveryDayOfMonth==true)
              loginEveryDayOfWeek = true
            else{
              if(TimeTools.getDayOfWeek(record.createTime) >= dayOfMonth){
                loginEveryDayOfWeek = true
              }
            }
            (record.createTime,loginEveryDayOfMonth,loginEveryDayOfWeek,detectWeekOfMonth(s._2,dayInfo._1),detectThisWeek(s._2(dayInfo._1-1)),record.modular)
          }else{
            (record.createTime,false,false,false,false,record.modular)
          }
        }
      }else{
        val arr = new Array[Int](5)
        for(i<-0 to dayInfo._1-2){
          arr(i) = 3
        }
        arr(dayInfo._1-1) = 1
        state.update((loginT,arr,dayInfo._2))
        (record.createTime,true,true,false,false,record.modular)
      }
    }

    val resultStream = loginStream.mapWithState(StateSpec.function(mapfun))
    resultStream.foreachRDD(rdd=>rdd.foreachPartition(it=>{
      val week = TimeTools.getWeek()
      val month = TimeTools.getMonth()
      var weekC = 0
      var monthC = 0
      val loginContinueOfMonth = new mutable.HashMap[Int,Int]()
      val loginContinueOfWeek = new mutable.HashMap[Int,Int]()
      it.foreach(it=>{//base on event time to process
        println("start.......")
        val t = TimeTools.formatTs(it._1)
        if(it._3) {
          loginContinueOfWeek.put(t, loginContinueOfWeek.getOrElse(t, 0) + 1)
        }
        if(it._2)
          loginContinueOfMonth.put(t,loginContinueOfMonth.getOrElse(t,0)+1)
        if(it._5)
          weekC = weekC + 1
        if(it._4)
          monthC = monthC + 1
      })
      val jedis = RedisOperation.getInstance(redisPro).getResource()
      val pl = jedis.pipelined()
      loginContinueOfWeek.foreach(entry=>{
        pl.hincrBy(RedisConfig.logLoginContinueWeekKey,entry._1.toString,entry._2)
      })
      loginContinueOfMonth.foreach(entry=>{
        pl.hincrBy(RedisConfig.logLoginContinueMonthKey,entry._1.toString,entry._2)
      })
      pl.hincrBy(RedisConfig.logLoginThreeMonthKey,month.toString,monthC)
      pl.hincrBy(RedisConfig.logLoginThreeWeekKey,week.toString,weekC)
      pl.sync()
      pl.close()
      jedis.close
    }))
    streaming.start()
    streaming.awaitTermination()
  }
}
